package idmp

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.graphx.{Edge, Graph, VertexId, VertexRDD}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import util.SparkUtil

/**
 * 三类埋点日志数据的id映射计算程序（没有考虑滚动实现，只有当天）
 */
object LogDataIDMapping {
  var flag = 0

  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkUtil.getSparkSession(this.getClass.getSimpleName)
    import spark.implicits._
    // 加载三类数据
    val appLog: Dataset[String] = spark.read.textFile("E:\\doit项目资料\\DW12-文档资料\\day03\\yiee_logs\\2020-01-11\\app")
    val webLog: Dataset[String] = spark.read.textFile("E:\\doit项目资料\\DW12-文档资料\\day03\\yiee_logs\\2020-01-11\\web")
    val wxAppLog: Dataset[String] = spark.read.textFile("E:\\doit项目资料\\DW12-文档资料\\day03\\yiee_logs\\2020-01-11\\wxApp")

    // 提取每一类数据中的标识字段id
    val app_ids: RDD[Array[String]] = extract(appLog)
    val web_ids: RDD[Array[String]] = extract(webLog)
    val wxApp_ids: RDD[Array[String]] = extract(wxAppLog)

    val ids: RDD[Array[String]] = app_ids.union(web_ids).union(wxApp_ids)

    // 提取点集合
    val vertices: RDD[(Long, String)] = ids.flatMap(line => line.map(a => (a.hashCode.toLong, a)))
    // 提取边集合
    // val edges: RDD[Edge[String]] = ids.flatMap(line => for (i <- 0 to line.length - 2) yield Edge(line(i).hashCode.toLong, line(i + 1).hashCode.toLong, ""))
    // 用双层for循环来对一个数组中的所有标识进行两两组合成边
    val edgesTmp: RDD[Edge[String]] = ids.flatMap(line => for (i <- 0 to line.length - 2; j <- i + 1 until line.length) yield Edge(line(i).hashCode.toLong, line(j).hashCode.toLong, ""))
    // 使用WC的方法过滤调出现次数小于经验阈值的边
    val edges: RDD[Edge[String]] = edgesTmp.map((_, 1)).reduceByKey(_ + _).filter(_._2 > 2).map(_._1)

    // 用构造好的点集合和边集合，构造一张图， 并调用图的最大连通子图算法
    val graph: Graph[String, String] = Graph(vertices, edges)
    val graph2: Graph[VertexId, String] = graph.connectedComponents()
    // 从结果图中取出图的点集合，即可以得到我们想要的分组结果
    val res_tuples: VertexRDD[VertexId] = graph2.vertices // VertexRDD[VertexId] ==> RDD[(Long, Long)]  (点id，组中最小值)
    // 可以直接使用图计算产生结果的组最小值，作为这一组的guid，也可以自己生成一个uuid进行相关转换操作作为guid
    res_tuples.toDF("biaoshi_hashcode", "guid").write.parquet("data/idmp/2020-01-11")
    spark.close()
  }

  // 提取日志数据DS中每一行的标识字段（如果不同类的日志数据中格式以及字段不完全相同则不能提取该方法）
  def extract(logDS: Dataset[String]): RDD[Array[String]] = {
    logDS.rdd.map(line => {
      // 将一行数据解析成json对象
      val jsonObj: JSONObject = JSON.parseObject(line)
      val userObj: JSONObject = jsonObj.getJSONObject("user")
      val uid = userObj.getString("uid")
      val phoneObj: JSONObject = userObj.getJSONObject("phone")
      val imei: String = phoneObj.getString("imei")
      val mac: String = phoneObj.getString("mac")
      val imsi: String = phoneObj.getString("imsi")
      val androidId: String = phoneObj.getString("androidId")
      val deviceId: String = phoneObj.getString("deviceId")
      val uuid: String = phoneObj.getString("uuid")
      Array(uid, imei, mac, imsi, androidId, deviceId, uuid).filter(StringUtils.isNotBlank(_))
    })
  }
}
